package top.jfunc.common.router;

import java.io.UnsupportedEncodingException;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.List;
import java.util.SortedMap;
import java.util.TreeMap;

/**
 * <a href="https://baijiahao.baidu.com/s?id=1709299096874847737&wfr=spider&for=pc">分布式理论——Consistent Hash（一致性哈希算法）</a>
 *
 * 一致性hash算法选取一个
 * a、virtual node：解决不均衡问题
 * b、hash method replace hashCode：String的hashCode可能重复，需要进一步扩大hashCode的取值范围
 */
public class ConsistentHashRouter<T,O> implements Router<T,O> {

    private static final int DEFAULT_VIRTUAL_NODE_NUM = 100;
    private int virtualNodeNum = DEFAULT_VIRTUAL_NODE_NUM;

    public ConsistentHashRouter(int virtualNodeNum) {
        this.virtualNodeNum = virtualNodeNum;
    }

    public ConsistentHashRouter() {
    }

    @Override
    public T select(List<T> toSelectedList, O otherParam) {
        SortedMap<Long, T> ring = new TreeMap<>();
        for (T t: toSelectedList) {
            for (int i = 0; i < virtualNodeNum; i++) {
                long hash = hash("SHARD-" + t + "-NODE-" + i);
                ring.put(hash, t);
            }
        }

        return ring.get(ring.firstKey());
        //return ring.get(ring.lastKey());
    }

    /**
     * get hash code on 2^32 ring (md5散列的方式计算hash值)
     */
    protected long hash(String key) {

        // md5 byte
        MessageDigest md5;
        try {
            md5 = MessageDigest.getInstance("MD5");
        } catch (NoSuchAlgorithmException e) {
            throw new RuntimeException("MD5 not supported", e);
        }
        md5.reset();
        byte[] keyBytes = null;
        try {
            keyBytes = key.getBytes("UTF-8");
        } catch (UnsupportedEncodingException e) {
            throw new RuntimeException("Unknown string :" + key, e);
        }

        md5.update(keyBytes);
        byte[] digest = md5.digest();

        // hash code, Truncate to 32-bits
        long hashCode = ((long) (digest[3] & 0xFF) << 24)
                | ((long) (digest[2] & 0xFF) << 16)
                | ((long) (digest[1] & 0xFF) << 8)
                | (digest[0] & 0xFF);

        return hashCode & 0xffffffffL;
    }

}
